package com.hxl.alert.consumer;

import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSON;
import com.hxl.alert.client.XxlJobApiClient;
import com.hxl.alert.constant.CacheConstants;
import com.hxl.alert.entity.AlertOriginBase;
import com.hxl.alert.entity.FailedAlertMessage;
import com.hxl.alert.entity.XxlJobInfo;
import com.hxl.alert.exception.AlertValidationException;
import com.hxl.alert.utils.AlertTimeUtils;
import com.hxl.alert.utils.RedisUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.spring.annotation.ConsumeMode;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.text.ParseException;
import java.time.LocalDateTime;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

@Component
@RocketMQMessageListener(
        topic = "ALERT_TOPIC",
        consumerGroup = "ALERT_CONSUMER_GROUP1",
        consumeMode = ConsumeMode.CONCURRENTLY,
        messageModel = MessageModel.CLUSTERING
)
public class AlertConsumer implements RocketMQListener<AlertOriginBase> {

    private final Logger logger = LoggerFactory.getLogger(getClass());

    @Resource
    private RedisUtils redisUtils;

    @Resource
    private XxlJobApiClient xxlJobClient;

    @Override
    public void onMessage(AlertOriginBase alert) {
        try {
            validateAlert(alert);
        } catch (AlertValidationException e) {
            e.printStackTrace();
            handleFailedAlert(alert, e.getMessage());
        }
        String faultId = alert.getFaultId();
        String lockKey = CacheConstants.PROCESS_LOCK + faultId;
        logger.info("[onMessage][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), alert);
        try {
            // 获取分布式锁
            Boolean locked = redisUtils.setIfAbsent(lockKey, "locked", 30, TimeUnit.SECONDS);

            if (Boolean.TRUE.equals(locked)) {
//                logger.info("获取锁成功");
                processAlert(alert);
            } else {
                // 稍后重试
                Thread.sleep(100);
                onMessage(alert);
            }
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            handleFailedAlert(alert, "系统错误: " + e.getMessage());
        } finally {
            redisUtils.delete(lockKey);
        }
    }

    private void processAlert(AlertOriginBase alert) {
//        logger.info("开始处理告警信息");
        try {
            // 获取当前时间戳
            String faultId = alert.getFaultId();
            long timestamp = AlertTimeUtils.toTimestamp(alert.getOccurTime());
            //先将所有告警存入redis
            handleReceiveMessage(alert);
            if ("越限".equals(alert.getFaultDescr())) {
                //存储数据后续方便计算
                redisUtils.zAdd(CacheConstants.YUEXIANAREA_KEY, alert, timestamp);
                handleYueXian(alert);
            } else {
                //存储数据后续方便计算
                redisUtils.zAdd(CacheConstants.FAULT_KEY, alert, timestamp);
                handleFault(alert);
            }
        } catch (ParseException e) {
            // 处理时间格式异常
            e.printStackTrace();
        }
    }

    private void handleReceiveMessage(AlertOriginBase alert) throws ParseException {
//        logger.info("开始处理接收到的告警信息");
        redisUtils.set(CacheConstants.RECEIVE_KEY + alert.getFaultId(), alert);
        redisUtils.zAdd(CacheConstants.SCORE_KEY, alert.getFaultId(), AlertTimeUtils.toTimestamp(alert.getOccurTime()));
    }

    private void handleYueXian(AlertOriginBase alert) throws ParseException {
        // 生成唯一任务ID（用于后续取消任务）
        String taskId = "limit_" + alert.getFaultId();
        // 存储越限告警上下文
        redisUtils.setIfAbsent(CacheConstants.YUEXIAN_KEY + taskId, alert, 60, TimeUnit.SECONDS);

        xxlJobClient.createJob(
                XxlJobInfo.builder()
                        .jobGroup(2)
//                        .scheduleType("CRON")
//                        .scheduleConf(getCronAfterOneMinute())
                        .scheduleType("FIX_RATE")   // 固定速率模式
                        .scheduleConf("1")      // 间隔1秒（单位：毫秒）
                        .executorHandler("alarmTimeoutHandler")
                        .executorParam(taskId)
                        .build()
        );
        logger.info("创建越限告警任务: {}", taskId);
    }

    private void handleFault(AlertOriginBase alert) throws ParseException {
        // 扫描所有未超时的越限告警
        Set<String> keys = redisUtils.scanKeys(CacheConstants.YUEXIAN_KEY + "limit_*");
        if (keys.isEmpty()) return;

        List<AlertOriginBase> limitAlerts = new ArrayList<>();
        keys.forEach(key -> {
            AlertOriginBase limitAlert = (AlertOriginBase) redisUtils.get(key);
            if (limitAlert != null) {
                String taskId = key.split(":")[2];
                xxlJobClient.findJobIdsAndStopJobByParam(taskId); // 根据参数终止任务
                redisUtils.delete(key);
            }
        });
        //TODO 混合计算逻辑 计算 接口返回越限区域+3分钟内越限的区域
        logger.info("开始混合计算逻辑");
        processHybrid(alert);
    }

    /**
     * 告警消息校验
     */
    private void validateAlert(AlertOriginBase alert) throws AlertValidationException {
        // 校验faultId
        if (StringUtils.isEmpty(alert.getFaultId())) {
            throw new AlertValidationException("faultId不能为空");
        }

        // 校验occurTime
        if (StringUtils.isEmpty(alert.getOccurTime())) {
            throw new AlertValidationException("occurTime不能为空");
        }

        try {
            AlertTimeUtils.toTimestamp(alert.getOccurTime());
        } catch (ParseException e) {
            throw new AlertValidationException("时间格式错误: " + alert.getOccurTime());
        }
    }

    /**
     * 处理失败消息
     */
    private void handleFailedAlert(AlertOriginBase alert, String reason) {
        FailedAlertMessage failedAlert = new FailedAlertMessage();
        failedAlert.setOriginalAlert(alert);
        failedAlert.setErrorReason(reason);
        failedAlert.setFailedTime(new Date());

        // 存储到Redis失败队列
        redisUtils.lRightPush(CacheConstants.FAILED_ALERTS_KEY, failedAlert);
        // 设置过期时间（7天）
        redisUtils.expire(CacheConstants.FAILED_ALERTS_KEY, 7, TimeUnit.DAYS);
        logger.error("消息处理失败: {}", reason);
    }

    private String getCronAfterOneMinute() {
        LocalDateTime time = LocalDateTime.now().plusMinutes(1);
        return String.format("0 %d %d * * ?", time.getMinute(), time.getHour());
    }

    //TODO 混合计算逻辑 计算 接口返回越限区域+3分钟内越限的区域
    private void processHybrid(AlertOriginBase faultAlert) {
        //一个设备故障可能在多个越限故障一分钟范围内 (后期要对越限区域去重)
        //根据设备故障时间往前推三分钟,如果碰到上一个设备故障则结束
        try {
            long faultTimestamp = AlertTimeUtils.toTimestamp(faultAlert.getOccurTime());
            long faultMinTimestamp = AlertTimeUtils.toTimestamp(faultAlert.getOccurTime()) - 180_000;
            List<AlertOriginBase> objects = redisUtils.zRangeByScore(CacheConstants.FAULT_KEY, faultMinTimestamp, faultTimestamp)
                    .stream().map(obj -> (AlertOriginBase) obj).collect(Collectors.toList());
            if (objects.size() >= 2) {
                faultMinTimestamp = AlertTimeUtils.toTimestamp(objects.get(objects.size() - 2).getOccurTime());
            }

            //计算越限区域
            List<AlertOriginBase> yueXianAreas = redisUtils.zRangeByScore(CacheConstants.YUEXIANAREA_KEY, faultMinTimestamp, faultTimestamp)
                    .stream().map(obj -> (AlertOriginBase) obj).collect(Collectors.toList());
            yueXianAreas.forEach(yueXianArea -> logger.info("触发混合计算: 越限ID={}, 故障ID={}", yueXianArea.getFaultId(), faultAlert.getFaultId()));
        } catch (ParseException e) {
            e.printStackTrace();
        }
    }
}